package com.powerbank.devicecenter.mqtt;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;

/**
 * MQTT消息通道配置
 */
@Slf4j
@Configuration
@RequiredArgsConstructor
public class MqttChannelConfig {

    private final MqttPahoClientFactory mqttClientFactory;
    private final MqttProperties mqttProperties;
    private final DeviceStatusMessageListener deviceStatusMessageListener;
    private final DeviceHeartbeatMessageListener deviceHeartbeatMessageListener;

    @Bean
    public IntegrationFlow mqttInboundFlow() {
        return IntegrationFlow.from(mqttInboundChannelAdapter())
                .handle(deviceStatusMessageListener)
                .get();
    }

    @Bean
    public IntegrationFlow mqttHeartbeatInboundFlow() {
        return IntegrationFlow.from(mqttHeartbeatInboundChannelAdapter())
                .handle(deviceHeartbeatMessageListener)
                .get();
    }

    private org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter mqttInboundChannelAdapter() {
        org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter adapter =
                new org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter(
                        mqttProperties.getBroker().getUrl(),
                        mqttProperties.getBroker().getClientId() + "-status",
                        mqttClientFactory,
                        mqttProperties.getTopics().getDeviceStatus()
                );
        adapter.setCompletionTimeout(5000);
        adapter.setQos(1);
        return adapter;
    }

    private org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter mqttHeartbeatInboundChannelAdapter() {
        org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter adapter =
                new org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter(
                        mqttProperties.getBroker().getUrl(),
                        mqttProperties.getBroker().getClientId() + "-heartbeat",
                        mqttClientFactory,
                        mqttProperties.getTopics().getDeviceHeartbeat()
                );
        adapter.setCompletionTimeout(5000);
        adapter.setQos(1);
        return adapter;
    }
}